package cn.wonhigh.o2o.openfire.plugins;

import java.io.File;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.dom4j.Element;
import org.jivesoftware.openfire.MessageRouter;
import org.jivesoftware.openfire.OfflineMessage;
import org.jivesoftware.openfire.OfflineMessageStore;
import org.jivesoftware.openfire.PresenceManager;
import org.jivesoftware.openfire.XMPPServer;
import org.jivesoftware.openfire.container.Plugin;
import org.jivesoftware.openfire.container.PluginManager;
import org.jivesoftware.openfire.interceptor.InterceptorManager;
import org.jivesoftware.openfire.interceptor.PacketInterceptor;
import org.jivesoftware.openfire.interceptor.PacketRejectedException;
import org.jivesoftware.openfire.session.Session;
import org.jivesoftware.openfire.user.UserManager;
import org.jivesoftware.util.PropertyEventDispatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xmpp.packet.JID;
import org.xmpp.packet.Message;
import org.xmpp.packet.Packet;
import org.xmpp.packet.Presence;

import cn.wonhigh.o2o.openfire.plugins.util.MessageReceiptCommand;
import cn.wonhigh.o2o.openfire.plugins.util.MessageReceiptConstants;

/**
 * 推送离线消息
 * 
 * @author wei.ss
 * @createDate 2015-03-09
 * @version 1.0
 */
public class OfflineMsgPlugin extends RedisInitializer implements
		PacketInterceptor, Plugin {
	
	private static final Logger log = LoggerFactory.getLogger(OfflineMsgPlugin.class);
	private final ScheduledExecutorService executorService = Executors.
			newScheduledThreadPool(MessageReceiptConstants.PROCESS_MESSAGE_THREAD_COUNT);

	private InterceptorManager interceptorManager;
	private OfflineMessageStore offlineMessageStore;
	private PresenceManager presenceManager;
	private UserManager userManager;
	private XMPPServer server;
	private MessageRouter messageRouter;

	public OfflineMsgPlugin() {
	}

	public void initializePlugin(PluginManager manager, File pluginDirectory) {
		PropertyEventDispatcher.addListener(this);

		interceptorManager = InterceptorManager.getInstance();
		interceptorManager.addInterceptor(this);

		server = XMPPServer.getInstance();
		offlineMessageStore = server.getOfflineMessageStore();
		userManager = server.getUserManager();
		presenceManager = server.getPresenceManager();
		messageRouter = server.getMessageRouter();
		
		// 初始化redis配置
		initRedis();

		log.info("初始化离线消息插件");
	}

	public void destroyPlugin() {
		interceptorManager.removeInterceptor(this);
		PropertyEventDispatcher.removeListener(this);
		
		// 关闭线程池
		shutdownAndAwaitTermination(executorService);
	}

	/**
	 * intercept message
	 */
	@Override
	public void interceptPacket(Packet packet, Session session,
			boolean incoming, boolean processed) throws PacketRejectedException {
		if(incoming && !processed){
			if(packet instanceof Message) {
				processRetry((Message)packet);
				return;
			}
			
			if (packet instanceof Presence) {
				Presence presence = (Presence) packet;
				JID from = presence.getFrom();
				if (!presence.isAvailable() || from == null) {
					log.info("离线插件，用户不是上线状态不做处理");
					return;
				}
				String username = from.getNode();
				if (username == null || !userManager.isRegisteredUser(from)) {
					log.info("离线插件，用户为空或者用户不存在：username={}", username);
					return;
				}
				this.processOfflineMessage(username);
			}
		}
	}
	
	private static String buildUserLoginKey(String username){
		String loginKey = "offline_msg_user_logined_"+username;
		return loginKey;
	}
	
	// 处理用户的离线消息
	private void processOfflineMessage(String username) {
		String loginKey = buildUserLoginKey(username);
		boolean needDeleteLoginKey = false;
		try {
			long exists = redisTemplate.setnx(loginKey, username);
			if(exists != 1) {
				String oldValue = redisTemplate.get(loginKey);
				log.info("用户已登录：username={}", oldValue);
				return;
			}
			needDeleteLoginKey = true;
			
			List<OfflineMessage> list = (List<OfflineMessage>)offlineMessageStore.getMessages(username, true);
			int totalCount = null == list ? 0 : list.size();
			log.info("offline message totalCount={}", totalCount);
			long begin = System.currentTimeMillis();
			if (totalCount > 0) {
				int pageSize = MessageReceiptConstants.PAGE_SIZE;
				int pageCount = (totalCount / pageSize);
				pageCount = totalCount % pageSize == 0 ? pageCount: (pageCount + 1);
				
				outer:
				for (int page = 0; page < pageCount; page++) {
					for(int i = 0; i < pageSize; i++) {
						int index = page * pageSize + i;
						if(index == totalCount) {
							break outer;
						}
						OfflineMessage msg = list.get(index);
						messageRouter.route(msg);
					}
					log.info("offline message page={}, pageCount={}",
							page, pageCount);
					// 每处理一页，暂停一会儿
					if (page + 1 < pageCount) {
						Thread.sleep(1000);
					}
				}
				log.info("处理离线消息话费时间：{}s", (System.currentTimeMillis() - begin) / 1000);
				log.info("用户（{}）状态在线，取得离线消息{}条", username, totalCount);
			}
		} catch (Exception e) {
			log.error("处理离线消息发生异常", e);
		} finally {
			if(needDeleteLoginKey){
				redisTemplate.del(loginKey);
			}
		}
	}
	
	public void processRetry(Message message){
		try {
			// 必要条件过滤 start
			if(null == message || message.getType() != Message.Type.chat){
				log.info("不是聊天消息，不做处理");
				return;
			}
			JID targetJID = message.getTo();
			if(targetJID == null || targetJID.getNode() == null){
				return;
			}
			String to = targetJID.getNode();
			if (targetJID.getDomain() == null || server.isRemote(targetJID)) {
				log.info("不是本域名下的消息：message={}", message);
				return;
			}
			// 必要条件过滤 end
			
			// 处理回执消息
			Element receiptElement = message.getChildElement("received", "urn:xmpp:receipts");
			if (receiptElement != null) { // 回执消息
				String messageId = receiptElement.attribute("id").getStringValue();
				String key = MessageReceiptConstants.getMessageIdKey(messageId);
				redisTemplate.set(key, messageId);
				log.info("回执消息：" + message.toXML());
				return;
			} else if(message.getBody() == null){
				log.info("不是聊天消息，不做处理");
				return;
			}
			if(null == to || to.length() > 20) {
				// 发送给微信的消息不做处理。由插件MessagePlugin处理
				return;
			}
			if (!userManager.isRegisteredUser(to)) {
				// receiver is requesting presence information of an anonymous user
				log.info("未注册用户：username={}", to);
				return;
			}
			
			// 设置接收端需要发送回执
			Element requestElement = message.getChildElement("request", "urn:xmpp:receipts");
			if(null == requestElement){
				message.addChildElement("request", "urn:xmpp:receipts");  
			}
			
			log.info("加入定时任务:message={}", message.toXML());
			executorService.schedule(new MessageReceiptCommand(messageRouter, 
					presenceManager, offlineMessageStore, userManager, redisTemplate, message), 
					MessageReceiptConstants.MESSAGE_TIMEOUT_INTERVAL, TimeUnit.SECONDS);
		} catch (Exception e) {
			log.warn("处理重试消息发生异常", e);
		}
	}
	
	private void shutdownAndAwaitTermination(ExecutorService pool) {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)){
                pool.shutdownNow();
                if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                    log.error("Pool did not terminate");
                }
            }
        }
        catch (InterruptedException ie) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}
